package com.dys.flink.wc2;

import com.dys.flink.wc.util.JsonUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.StateBackendOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Socket 实时流，按 user_id 分组，内部按 product_name 累计 count & totalPrice
 * 运行前：nc -lk 9999
 */
public class UserProductStatJob {

    public static void main(String[] args) throws Exception {
        Configuration config = new Configuration();
        config.set(StateBackendOptions.STATE_BACKEND, "hashmap");
        try (StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config)) {
            env.setParallelism(1);

            env.socketTextStream("localhost", 9999)
                    .filter(StringUtils::isNotEmpty)
                    .map(line -> JsonUtils.fromJson(line, Order.class))
                    .returns(Order.class)
                    .keyBy(o -> o.user_id)
                    .process(new UserProductProc())
                    .print();

            env.execute("UserProductStatJob");
        }
    }

    // ---------------- 处理函数 ----------------
    public static class UserProductProc extends KeyedProcessFunction<String, Order, String> {

        // Map<product_name, ProductStat>
        private MapState<String, ProductStat> mapState;

        @Override
        public void open(Configuration parameters) {
            MapStateDescriptor<String, ProductStat> descriptor = new MapStateDescriptor<>("user-product-stat", TypeInformation.of(String.class), TypeInformation.of(ProductStat.class));
            mapState = getRuntimeContext().getMapState(descriptor);
        }

        @Override
        public void processElement(Order o, Context ctx, Collector<String> out)
                throws Exception {

            ProductStat stat = mapState.get(o.product_name);
            if (stat == null) {
                stat = new ProductStat(o.product_name, 0L, 0.0);
            }
            stat.count += 1;
            stat.totalPrice += o.price;
            mapState.put(o.product_name, stat);

            // 输出当前用户的所有产品汇总
            StringBuilder sb = new StringBuilder();
            sb.append("userId=").append(o.user_id).append(" -> ");
            for (var entry : mapState.entries()) {
                sb.append(entry.getValue()).append(" | ");
            }
            out.collect(sb.toString());
        }
    }

    // ---------------- POJO ----------------
    public static class Order {
        public long id;
        public String user_id;
        public String product_name;
        public String product_id;
        public double price;
    }

    public static class ProductStat {
        public String productName;
        public long count;
        public double totalPrice;

        public ProductStat() {}

        public ProductStat(String productName, long count, double totalPrice) {
            this.productName = productName;
            this.count = count;
            this.totalPrice = totalPrice;
        }

        @Override
        public String toString() {
            return String.format("%s[count=%d, total=%.2f]", productName, count, totalPrice);
        }
    }
}